package cn.ac.iie.di.ban.data.exchange.runner.slave.task.runTask;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.StatusEnum;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;

import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.SendMQ.SendMQ;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.data.RocketMQArgvData;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.receiveMQ.ReceiveMQ;
import cn.ac.iie.di.ban.data.exchange.runner.slave.SubscriptionSlave;
import cn.iie.jsqlparser.helper.JSQLParserHelper;
import cn.iie.jsqlparser.helper.SQLOperatorCompare;
import cn.iie.jsqlparser.helper.data.CondTreeData;
import cn.iie.jsqlparser.helper.data.CondTreeValueData;
import cn.iie.jsqlparser.helper.data.SQLOperatorEnum;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by Black-Laptop on 12/6/2017.
 */
public class SubscriptionTaskThread extends Thread {

    private static final Logger LOG = LoggerFactory.getLogger(SubscriptionTaskThread.class);

    private Boolean stopThread;
    private RocketMQArgvData rocketMQArgvData;
    private ArrayBlockingQueue<REMessage> queue;
    private SendMQ sendMQ;
    private JSQLParserHelper jsqlParserHelper;

    private String sql;
    private List<String> selectItems;
    private CondTreeData condTree;

    public SubscriptionTaskThread(JSQLParserHelper jsqlParserHelper, RocketMQArgvData rocketMQArgvData) throws Exception {
        this.stopThread = false;
        this.jsqlParserHelper = jsqlParserHelper;
        this.rocketMQArgvData = rocketMQArgvData;
        this.sendMQ = null;
        this.sql = jsqlParserHelper.getSQLString();
        init();
    }

    @Override
    public void run() {
        LOG.info("sql:" + sql);
        REMessage message = null;
        while (!stopThread) {
            try {
                message = queue.poll(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                SubscriptionSlave.callAfterFinished.finished(StatusEnum.SUBSCRIPTION_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
                LOG.error(ExceptionUtils.getFullStackTrace(e));
            }
            if (message != null) {
                getAllowMessage(message);
                REMessage resultMessage = getAllowMessage(message);

                if (resultMessage == null) {
                    continue;
                }

                try {
                    sendMQ.sendMessage(resultMessage);
                } catch (Exception e) {
                    SubscriptionSlave.callAfterFinished.finished(StatusEnum.SUBSCRIPTION_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
                    LOG.error(ExceptionUtils.getFullStackTrace(e));
                }
            }
        }
        SubscriptionSlave.callAfterFinished.finished(StatusEnum.SUBSCRIPTION_TASK_SUCCESS, sql);
    }

    public void stopThread() {
        this.stopThread = true;
    }

    private void init() throws Exception {
        queue = new ArrayBlockingQueue<>(5);
        this.selectItems = jsqlParserHelper.getSQLSelectItems();
        this.condTree = jsqlParserHelper.getSQLWhere();

        startReceiveMQ();
    }

    private void startReceiveMQ() throws Exception {
        try {
            ReceiveMQ receiveMQ = new ReceiveMQ(this.rocketMQArgvData, queue);
            receiveMQ.start();
        } catch (REConnectionException | ParseException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Exception("start rocketMQ error");
        }
    }

    private void startSendMQ(REMessage message) throws Exception {
        Boolean sendMQIsNull = this.sendMQ == null;
        if (sendMQIsNull) {
            sendMQ = new SendMQ(message, this.rocketMQArgvData);
            sendMQ.start();

        }
    }

    private REMessage getAllowMessage(REMessage message) {
        REMessage allowMessage = null;
        try {
            allowMessage = ScreenAllowMessage(message);
        } catch (Exception e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
        }

        if (allowMessage == null || allowMessage.data.size() == 0) {
            return null;
        }

        REMessage resultMessage = getResultMessage(allowMessage);

        try {
            startSendMQ(resultMessage);
        } catch (Exception e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Error("start send mq error.");
        }

        return resultMessage;
    }

    private REMessage ScreenAllowMessage(REMessage message) throws Exception {
        List<String> colNames = Arrays.asList(message.colNames);
        List<Object[]> datas = message.data;
        List<Object[]> resultDatas = new ArrayList<>();

        for (Object[] data : datas) {
            Boolean dataAllow = checkDataAllow(colNames, data, this.condTree);
            if (dataAllow) {
                resultDatas.add(data);
            }
        }

        if (datas.size() < 1) {
            return null;
        }

        REMessage allowMessage = new REMessage(message.colNames, message.types, resultDatas, message.docSchemaName, message.userDesc);

        return allowMessage;
    }

    private Boolean checkDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        Boolean isLastChild = whereTree.getLeftChild() == null && whereTree.getRightChild() == null;
        if (isLastChild) {
            Boolean compareResult = checkChildDataAllow(colNames, data, whereTree);
            return compareResult;
        }

        Boolean compareResult = checkFatherDataAllow(colNames, data, whereTree);
        return compareResult;
    }

    private Boolean checkChildDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        CondTreeValueData valueData = whereTree.getValue();
        String compareKey = valueData.getKey();
        String compareValue = valueData.getValue();
        SQLOperatorEnum compareOperator = valueData.getOperator();

        int colIndex = colNames.indexOf(compareKey);
        Object value = data[colIndex];

        Boolean compareResult = SQLOperatorCompare.SQLOperatorCompare(compareOperator, value, compareValue);

        return compareResult;
    }

    private Boolean checkFatherDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        CondTreeData leftChild = whereTree.getLeftChild();
        CondTreeData rightChild = whereTree.getRightChild();
        SQLOperatorEnum compareOperator = whereTree.getValue().getOperator();
        Boolean leftChildResult = null;
        Boolean rightChildResult = null;
        if (leftChild != null) {
            leftChildResult = checkDataAllow(colNames, data, leftChild);
        }
        if (rightChild != null) {
            rightChildResult = checkDataAllow(colNames, data, rightChild);
        }

        if (leftChild == null) {
            return rightChildResult;
        } else if (rightChild == null) {
            return leftChildResult;
        }
        Boolean compareResult = SQLOperatorCompare.SQLOperatorCompare(compareOperator, leftChildResult, rightChildResult);

        return compareResult;
    }

    private REMessage getResultMessage(REMessage message) {
        List<Object[]> datas = message.data;
        List<String> colNames = Arrays.asList(message.colNames);
        List<REFieldType> types = Arrays.asList(message.types);

        List<Object[]> datasResult = getResultDatas(message);
        String[] colNamesResult = getResultColumns();
        REFieldType[] typesResult = getResultTypes(message);

        return new REMessage(colNamesResult, typesResult, datasResult, message.docSchemaName, message.userDesc);
    }

    private List<Object[]> getResultDatas(REMessage message) {
        List<Object[]> datas = message.data;
        List<String> colNames = Arrays.asList(message.colNames);

        List<Object[]> datasResult = new ArrayList<>();

        for (Object[] data : datas) {
            List<Object> dataResult = new ArrayList<>();

            for (String column : selectItems) {
                int columnIndex = colNames.indexOf(column);
                dataResult.add(data[columnIndex]);
            }

            datasResult.add(dataResult.toArray(new Object[dataResult.size()]));
        }

        return datasResult;
    }

    private String[] getResultColumns() {
        return selectItems.toArray(new String[selectItems.size()]);
    }

    private REFieldType[] getResultTypes(REMessage message) {
        List<String> colNames = Arrays.asList(message.colNames);
        REFieldType[] types = message.types;

        List<REFieldType> typesResult = new ArrayList<>();

        for (String column : selectItems) {
            int columnIndex = colNames.indexOf(column);
            typesResult.add(types[columnIndex]);
        }

        return typesResult.toArray(new REFieldType[typesResult.size()]);
    }
}
